package tjs.study.notes.dotest.io.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
 * Created by Nicky.Tang on 2021/11/2 4:28 下午
 *
 * @since 02.12.10
 */
public class DoTestOfServer {
    static String host = "127.0.0.1";
    static int port = 8081;
    static int clientNum = 1000;

    public static void main(String[] args) throws Exception {
        // 打开一个通道
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        // 开启非阻塞模式(阻塞模式就不是这种写法了，Selector必须要在非阻塞模式下运行，否则会报错)
        serverChannel.configureBlocking(false);

        // 开启选择器
        Selector selector = Selector.open();

        // 注册感兴趣的事件
        // 连接就绪 仅客户端有该事件
        //serverChannel.register(selector, SelectionKey.OP_CONNECT);
        // 接收就绪 仅服务端有该事件
        //serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        // 读就绪  就绪条件为：客户端有数据发送过来、客户端被关闭了等
        //serverChannel.register(selector, SelectionKey.OP_READ);
        // 写就绪  就绪条件为：操作系统内核缓冲区有空闲空间（不应该注册OP_WRITE事件，不然select()将一直有数据返回，导致占用资源）
        // serverChannel.register(selector, SelectionKey.OP_WRITE);
        // 同时注册多个事件
        SelectionKey serverSelect = serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        serverSelect.attach("服务端主选择器");

        // 绑定连接
        serverChannel.socket().bind(new InetSocketAddress(host, port));
        System.out.println("服务端已启动");

        while (selector.select() > 0) {//select会阻塞，直到至少一个事件就绪
            // 获取本次select轮询时的结果 -> 已就绪的事件集合
            Iterator<SelectionKey> selectionIterator = selector.selectedKeys().iterator();
            while (selectionIterator.hasNext()) {
                SelectionKey curSelection = selectionIterator.next();

                try {
                    // 下面开始根据不同的事件处理不同的业务
                    if (curSelection.isAcceptable()) {
                        // 当前为服务A,服务B连过来了，则要为服务A和B建立一个通道，并加入到selector监听中
                        // 这样服务A就可以监听服务B发过来的信息了，也可以主动发消息往服务B
                        registerOneClient(serverChannel, selector, curSelection);
                    } else if (curSelection.isReadable()) {
                        // 获取当前选择器的通道curSelection.channel()
                        SocketChannel curChannel = (SocketChannel) curSelection.channel();
                        // 开始读取数据（此处有很多坑，半包问题等）
                        String params = redaData(curSelection, curChannel);
                        responseData(curChannel, params);
                    }
                } finally {
                    // 从选择器中移除当前Selection，否则下次轮询依然是已就绪的状态
                    selectionIterator.remove();
                }
            }

        }
    }

    private static void registerOneClient(ServerSocketChannel serverChannel, Selector selector, SelectionKey curSelection) throws IOException {
        System.out.println(curSelection.channel() == serverChannel);//true
        // 获取当前选择器的通道curSelection.channel()
        SocketChannel clientChannel = ((ServerSocketChannel) curSelection.channel()).accept();
        clientChannel.configureBlocking(false);
        SelectionKey clientSelect = clientChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
        clientSelect.attach("客户端选择器-" + clientNum++);
    }

    public static String redaData(SelectionKey curSelection, SocketChannel curChannel) throws IOException {
        String params = null;
        ByteBuffer readBuffer = ByteBuffer.allocate(1024);// HeapByteBuffer
        int readBytes = curChannel.read(readBuffer);
        if (readBytes > 0) {
            // 反转读写模式
            readBuffer.flip();
            byte[] bytes = new byte[readBuffer.remaining()];
            readBuffer.get(bytes);
            params = new String(bytes, "UTF-8");
            System.out.println("来自客户端[" + curSelection.attachment() + "]的输入： [" + params.trim() + "]！");
        } else if (readBytes < 0) {
            // 客户端链接关闭时会触发OP_READ事件
            curSelection.cancel();
            curChannel.close();
        }
        return params;
    }

    public static void responseData(SocketChannel curChannel, String params) throws IOException {
        // 这里做一些业务操作
        String response = "来自服务器端的响应：" + params;

        try {
            Thread.currentThread().sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 响应业务操作的结果给客户端
        writeData(curChannel, response);
    }

    public static void writeData(SocketChannel channel, String data) throws IOException {
        if (data != null && data.trim().length() > 0) {
            byte[] bytes = data.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocateDirect(bytes.length);// DirectByteBuffer
            writeBuffer.put(bytes);
            writeBuffer.flip();
            channel.write(writeBuffer);
        }
    }

}
